package com.lzq.mq.nameserver.event.spi.listener;

import com.alibaba.fastjson2.JSON;
import com.lzq.mq.common.coder.TcpMsg;
import com.lzq.mq.common.dto.PullBrokerIpRespDTO;
import com.lzq.mq.common.enums.BrokerRegistryRoleEnum;
import com.lzq.mq.common.enums.NameServerResponseCode;
import com.lzq.mq.common.enums.RegistryTypeEnum;
import com.lzq.mq.common.event.Listener;
import com.lzq.mq.nameserver.common.CommonCache;
import com.lzq.mq.nameserver.event.model.PullBrokerIpEvent;
import com.lzq.mq.nameserver.store.ServiceInstance;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;

import static com.lzq.mq.common.constants.BrokerConstants.*;

/**
 * 拉取broker节点的监听器
 */
public class PullBrokerIpListener implements Listener<PullBrokerIpEvent> {

    /**
     * 处理 PULL_BROKER_IP_LIST请求
     * 请求参数: PullBrokerIpReqDTO
     * 返回参数: PullBrokerIpRespDTO
     * 必填参数 : Broker组 group
     * 选填参数 : 角色role(不填获取Broker组的所有节点)
     * @param event 事件对象
     */
    @Override
    public void onReceive(PullBrokerIpEvent event) {
        String pullRole = event.getRole();
        String pullGroup = event.getGroup();
        PullBrokerIpRespDTO pullBrokerIpRespDTO = new PullBrokerIpRespDTO();
        List<String> addressList = new ArrayList<>();
        Map<String, ServiceInstance> serviceInstanceMap = CommonCache.getServiceInstanceManager().getServiceInstanceMap();
        for (String reqId : serviceInstanceMap.keySet()) {
            ServiceInstance serviceInstance = serviceInstanceMap.get(reqId);

            // 是broker, 并且是相同的消费组
            if(RegistryTypeEnum.BROKER.getCode().equals(serviceInstance.getRegistryType())
                    && pullGroup.equals(serviceInstance.getAttrs().getOrDefault(BROKER_GROUP_NAME_KEY,""))){

                Map<String,Object> brokerAttrs = serviceInstance.getAttrs();
                String role = (String) brokerAttrs.get(BROKER_ROLE_KEY);
                /*
                为什么这么判断?
                为了防止出现非法的角色?
                 */
                // 是或的关系, 满足任意一个条件就返回
                if (pullRole == null || pullRole.isEmpty()) {
                    addressList.add(serviceInstance.getIp()+":"+serviceInstance.getPort());
                } else if (BrokerRegistryRoleEnum.MASTER.getCode().equals(pullRole)
                        && BrokerRegistryRoleEnum.MASTER.getCode().equals(role)) {
                    addressList.add(serviceInstance.getIp()+":"+serviceInstance.getPort());
                } else if (BrokerRegistryRoleEnum.SLAVE.getCode().equals(pullRole)
                        && BrokerRegistryRoleEnum.SLAVE.getCode().equals(role)) {
                    addressList.add(serviceInstance.getIp()+":"+serviceInstance.getPort());
                } else if (BrokerRegistryRoleEnum.SINGLE.getCode().equals(pullRole)
                        && BrokerRegistryRoleEnum.SINGLE.getCode().equals(role)) {
                    addressList.add(serviceInstance.getIp()+":"+serviceInstance.getPort());
                }


            }
        }
        pullBrokerIpRespDTO.setMsgId(event.getMsgId());
        pullBrokerIpRespDTO.setAddressList(addressList);

        TcpMsg tcpMsg = new TcpMsg(
                NameServerResponseCode.PULL_BROKER_ADDRESS_SUCCESS.getCode(),
                JSON.toJSONBytes(pullBrokerIpRespDTO)
        );
        event.getChannelHandlerContext().writeAndFlush(tcpMsg);
    }
}